package com.asiainfo.dacp.datastash.broker.facade;

import com.asiainfo.dacp.datastash.broker.core.BrokerContext;
import com.asiainfo.dacp.datastash.broker.core.StashUnitProcessModel;
import com.asiainfo.dacp.datastash.broker.core.enums.UnitTaskStateEnum;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitTask;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitTaskHeartbeat;
import com.asiainfo.dacp.datastash.broker.facade.state.UnitTaskStateLogService;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogVo;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/**
 *
 * 持续规划分片任务 当一个任务分片数大于处理数时，需要对这个任务持续的进行分片规划操作
 *
 * @author MeiKefu
 * @date 2017-12-1
 */
@Service
public class UnitTaskFacade {

	@Autowired
	private StashUnitProcessModel stashUnitProcessModel;

	@Autowired
	private UnitTaskStateLogService unitTaskStateLog;

	@Autowired
	private StashTraceLogger tracelog;
	private String clzName = this.getClass().getSimpleName();

	public StashUnitTask fetchWaitingTask() {
		return stashUnitProcessModel.fetchUnitTaskWaitQueue();
	}

	public StashUnitTask fetchDispatchTask() {
		return stashUnitProcessModel.fetchUnitTaskDispatchQueue();
	}

	public boolean isUnitTaskSuccess(String taskProcId) {
		String state = stashUnitProcessModel.getUnitTaskState(taskProcId);
		return UnitTaskStateEnum.SUCESS.value().equals(state);
	}

	public void dispatchTask(StashUnitTask stashUnitTask) {
		stashUnitProcessModel.pushUnitTaskDispatchQueue(stashUnitTask);
	}

	public void reQueueTask(StashUnitTask stashUnitTask) {
		stashUnitProcessModel.pushUnitTaskWaitQueue(stashUnitTask);
	}

	public void taskProcess(StashUnitTask stashUnitTask) {
		stashUnitTask.running();
		tracelog.info(StashTraceLogVo.buildStashTraceLogVo(stashUnitTask, clzName, "taskProcess", "task处理开始"));

		unitTaskStateLog.update(stashUnitTask.getTaskProcId(), stashUnitTask.getUnitProcId(),
				stashUnitTask.getUnitCode(), stashUnitTask.getBatchNo(), UnitTaskStateEnum.RUNNING.value(),
				"task开始执行", stashUnitTask.getBrokerId(), BrokerContext.getInstance().getBrokerIp());
	}

	public void taskComplete(StashUnitTask stashUnitTask) {
		stashUnitTask.success();
		tracelog.info(StashTraceLogVo.buildStashTraceLogVo(stashUnitTask, clzName, "taskComplete", "task处理完成"));

		unitTaskStateLog.update(stashUnitTask, UnitTaskStateEnum.SUCESS, "task执行完成",
				BrokerContext.getInstance().getBrokerId(), BrokerContext.getInstance().getBrokerIp());
	}

	public void taskFailed(StashUnitTask stashUnitTask) {
		stashUnitTask.failed();
		tracelog.info(StashTraceLogVo.buildStashTraceLogVo(stashUnitTask, clzName, "taskFailed", "task处理失败"));

		unitTaskStateLog.update(stashUnitTask, UnitTaskStateEnum.FAIL, "task执行失败",
				BrokerContext.getInstance().getBrokerId(), BrokerContext.getInstance().getBrokerIp());
	}

	/**
	 * 维持task心跳
	 */
	public void taskHearbeat(StashUnitTaskHeartbeat taskHeartbeat) {
		// 汇报task状态，判断task是否存活
		stashUnitProcessModel.taskHeartbeat(taskHeartbeat);
	}
}
